package com.shujia.streaming

import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Durations, StreamingContext}

object Demo02UpdateStateByKey {
  def main(args: Array[String]): Unit = {
    // 如果每个批次计算结果需要结合上一个批次的结果 一起考虑 则需要使用有状态算子
    val conf: SparkConf = new SparkConf()
    conf.setAppName("Demo02UpdateStateByKey")
    conf.setMaster("local[2]")

    val ssc: StreamingContext = new StreamingContext(new SparkContext(conf), Durations.seconds(5))
    ssc.checkpoint("spark/data/ck")

    val lineDS: DStream[String] = ssc.socketTextStream("master", 8888)

    lineDS
      .flatMap(_.split(","))
      .map(word => (word, 1))

      /**
       * 使用有状态算子需要设置Checkpoint的目录
       * 需要接收一个函数f:
       * 函数f的参数有两个分别是：
       * Seq类型（当前批次的同一个Key对应的所有Value）
       * Option类型（对于同一个Key来说，上一个批次中该Key的结果）
       */
      .updateStateByKey((seq: Seq[Int], opt: Option[Int]) => {
        // 当前批次中Key对应的所有Value之和
        val sum: Int = seq.sum
        val last_value: Int = opt match {
          case Some(v) =>
            v
          case None =>
            0
        }
        Some(sum + last_value)
      })
      .print()


    ssc.start()
    ssc.awaitTermination()
    ssc.stop()

  }

}
